package demo.db;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.SQLFeatureNotSupportedException;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.sql.DataSource;

import com.mongodb.client.MongoDatabase;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;

import demo.java.util.concurrent.ThreadFactoryDemo.NamedThreadFactory;

/**
 * 一个用于监控REDIS、mongo、mysql链接可用性的监控服务，建议设置5分钟执行一次。
 * <p>
 * 强依赖Spring。通过Spring上下文获取org.springframework.data.redis.core.RedisTemplate、
 * org.springframework.data.mongodb.core.MongoTemplate、javax.sql.DataSource对象来进行有效性校验。
 * 
 * @author hanjy
 *
 */
public class DbMonitor implements ApplicationContextAware {

    private static final Logger logger = LoggerFactory.getLogger(DbMonitor.class);

    /** Spring应用上下文 */
    private ApplicationContext applicationContext;
    /** 启用CAT报警 */
    private boolean enableCAT;
    /** 启用 testDB MQ报警 */
    private boolean enableTestDB;
    /** 校验mysql数据源 */
    private boolean checkDS;
    /** 校验mongo */
    private boolean checkMongo;
    /** 校验REDIS */
    private boolean checkRedis;

    private static final String MSG_FORMAT = "ping %s server [ %s ] not ok.";

    private ThreadPoolExecutor threadPool = null;
    private Map<Future<?>, Long> map = null;

    private String projectId;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    /**
     *
     * @param application
     * @param enableCAT
     * @param hasDS
     * @param hasMongo
     * @param hasRedis
     */
    public DbMonitor(ApplicationContext application, boolean enableCAT, boolean hasDS, boolean hasMongo,
                     boolean hasRedis) {
        if (application == null) {
            throw new IllegalArgumentException("ApplicationContext is null.");
        }
        if (enableTestDB && projectId == null) {
            throw new IllegalArgumentException("projectId is null,can't enable testDB alarm.");
        }
        this.applicationContext = application;
        this.enableCAT = enableCAT;
        this.checkMongo = hasMongo;
        this.checkDS = hasDS;
        this.checkRedis = hasRedis;
        if (hasDS) {
            Map<String, DataSource> dsMap = getBeansOfType(DataSource.class);
            int count = (dsMap == null || dsMap.isEmpty()) ? 0 : dsMap.size();
            logger.info("found {} DataSource Object", count);
            if (count > 0) {
                this.threadPool = new ThreadPoolExecutor(1, count, 0L, TimeUnit.MILLISECONDS,
                        new LinkedBlockingQueue<Runnable>(), new NamedThreadFactory("checkDS"));
                this.map = new LinkedHashMap<>(count);
            }
        }
    }

    /**
     * 发送CAT消息
     * 
     * @param errMsg
     * @param errMsg
     */
    private void sendCatInfo(String errMsg) {
        if (enableCAT) {
            logger.info("sendCatInfo:{}", errMsg);
        }
    }

    /**
     * alarm info 告警类型 实时告警)
     *
     * @param detail
     *            明细
     * @param remark
     *            备注
     */
    private void sendAlarmInfo(String detail, String remark) {
        if (enableTestDB) {
            logger.info("sendAlarmInfo:{}", detail);
        }
    }

    private void checkDataSource(String name, DataSource dataSource) {
        if (map.size() > 0) {
            for (Iterator<Map.Entry<Future<?>, Long>> iterator = map.entrySet().iterator(); iterator.hasNext();) {
                Map.Entry<Future<?>, Long> entry = iterator.next();
                Future<?> future = entry.getKey();
                if (future.isDone()) {
                    iterator.remove();
                } else {
                    long cost = System.currentTimeMillis() - entry.getValue();
                    if (cost > 20_000) {
                        String msg = "ping mysql datasource timeout:[20s]";
                        logger.error(msg);
                        sendCatInfo(msg);
                        sendAlarmInfo(msg, null);
                    }
                }
            }
        }

        if (map.size() >= threadPool.getMaximumPoolSize()) {
            logger.info("we had {} threads to checkDataSource,no more create thread.", map.size());
            return;
        }
        Future<?> future = threadPool.submit(() -> {
            String url = name;
            try (Connection connection = dataSource.getConnection()) {
                DatabaseMetaData metaData = connection.getMetaData();
                url = metaData.getURL();
                boolean isValid = connection.isValid(10);
                if (isValid) {
                    logger.info("ping mysql {} ok.", url);
                } else {
                    logger.info("ping mysql {} not ok.", url);
                    String msg = String.format(MSG_FORMAT, "mysql", url);
                    sendCatInfo(msg);
                    sendAlarmInfo(msg, null);
                }
            } catch (SQLFeatureNotSupportedException nse) {
                logger.warn("数据源[{}]不支持[{}]操作，不做有效性校验", name, nse.getMessage());
            } catch (Exception e) {
                logger.error("校验MySQL数据源异常：{}", e.getMessage());
                String msg = String.format(MSG_FORMAT, "mysql", url);
                sendCatInfo(msg);
                sendAlarmInfo(msg, null);
            }
        });
        map.put(future, System.currentTimeMillis());
    }

    private void checkMongoTemplate2(MongoTemplate mongoTemplate) {
        String serverAddress = null;
        try {
            MongoDatabase db = mongoTemplate.getDb();
            serverAddress = db.getName();

        } catch (Exception e) {
            logger.error("校验Mongo异常：{}", e.getMessage());
            String msg = String.format(MSG_FORMAT, "mongo", serverAddress);
            sendCatInfo(msg);
            sendAlarmInfo(msg, null);
        }
    }

    private void checkRedisTemplate(RedisTemplate redisTemplate) {
        String serverAddress = "";
        String msg = null;
        RedisConnection redisConnection = null;
        try {
            RedisConnectionFactory redisConnectionFactory = redisTemplate.getConnectionFactory();
            JedisConnectionFactory jedisConnectionFactory = (JedisConnectionFactory) redisConnectionFactory;
            String hostName = jedisConnectionFactory.getHostName();
            int port = jedisConnectionFactory.getPort();
            serverAddress = hostName + ":" + port;
            redisConnection = redisConnectionFactory.getConnection();
            String pong = redisConnection.ping();
            if ("PONG".equalsIgnoreCase(pong)) {
                logger.info("ping redis {} ok.", serverAddress);
            } else {
                logger.error("ping redis {} not ok.", serverAddress);
                msg = String.format(MSG_FORMAT, serverAddress);
                sendCatInfo(msg);
                sendAlarmInfo(msg, null);
            }
        } catch (Exception e) {
            logger.error("校验Redis异常：{}", e.getMessage());
            msg = String.format(MSG_FORMAT, serverAddress);
            sendCatInfo(msg);
            sendAlarmInfo(msg, null);
        } finally {
            if (redisConnection != null && !redisConnection.isClosed()) {
                redisConnection.close();
            }
        }

    }

    /**
     * 获取mysql、redis、mongo的连接，校验是否正常
     */
    public void checkConnection() {
        int count = 0;
        if (checkRedis) {
            Map<String, RedisTemplate> redises = getBeansOfType(RedisTemplate.class);
            count = (redises == null || redises.isEmpty()) ? 0 : redises.size();
            logger.info("获取{}个RedisTemplate对象", count);
            if (count > 0) {
                redises.values().forEach(this::checkRedisTemplate);
            }
        }
        if (checkMongo) {
            Map<String, MongoTemplate> mongos = getBeansOfType(MongoTemplate.class);
            count = (mongos == null || mongos.isEmpty()) ? 0 : mongos.size();
            logger.info("获取{}个MongoTemplate对象", count);
            if (count > 0) {
                mongos.values().forEach(this::checkMongoTemplate2);
            }
        }

        if (checkDS) {
            Map<String, DataSource> dsMap = getBeansOfType(DataSource.class);
            count = (dsMap == null || dsMap.isEmpty()) ? 0 : dsMap.size();
            logger.info("获取{}个DataSource对象", count);
            if (count > 0) {
                dsMap.forEach(this::checkDataSource);
            }
        }
    }

    private <T> Map<String, T> getBeansOfType(Class<T> type) {
        try {
            return applicationContext.getBeansOfType(type);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return null;
    }

    /**
     * 销毁对象
     */
    public void destroy() {
        if (threadPool != null && !threadPool.isShutdown()) {
            threadPool.shutdown();
        }
        if (map != null && !map.isEmpty()) {
            map.forEach((k, v) -> {
                k.cancel(true);
                map.remove(k);
            });
        }
    }

}
